from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 构建SparkSession对象
    spark = SparkSession.builder. \
        appName("local[*]"). \
        config("spark.sql.shuffle.partitions", "4"). \
        getOrCreate()
    # appName 设置程序名称
    # config 设置常用属性。可以通过此来设置配置
    # 最后通过getOrCreate 创建 SparkSession对象

    # 从SparkSession中获取SparkContext
    sc = spark.sparkContext

    # 1.获取数据集
    schema = StructType() \
        .add("user_id", StringType(), nullable=True) \
        .add("movie_id", IntegerType(), nullable=True) \
        .add("rank", IntegerType(), nullable=True) \
        .add("ts", StringType(), nullable=True)

    df = spark.read.format("csv") \
        .option("sep", "\t") \
        .option("header", "false") \
        .option("encoding", "utf-8") \
        .schema(schema) \
        .load("hdfs://node1:8020/input/u.data")

    # JDBC 写出数据
    df.write.format("jdbc").mode("overwrite") \
        .option("url", "jdbc:mysql://192.168.111.128:3306/spark?useSSL=false&useUnicode=true")\
        .option('dbtable', 'movies') \
        .option('user', 'root') \
        .option('password', '1234') \
        .save()
    # jdbc写出，会自动创建表
    # 因为DataFrame中有表结构信息，StructType记录有每列的名称，类型以及是否允许为空

    # JDBC 读取数据
    df2 = spark.read.format("jdbc") \
        .option("url", "jdbc:mysql://192.168.111.128:3306/spark?useSSL=false&useUnicode=true") \
        .option('dbtable', 'movies') \
        .option('user', 'root') \
        .option('password', '1234') \
        .load()

    df2.printSchema()
    df2.show(truncate=False)
